#include <unistd.h>
#include <sys/time.h>
#include <modbus/modbus.h>
#include <mosquitto.h>
#include "seqiot.h"
#include "channel.h"
#include "modbus_acquire.h"
#include "cJSON.h"

void sched_modbus_device_segment(channel_t *channel, data_catalog_e catalog, device_t *device, segment_e segment) {
    int i;
    int start = -1, prev = start, next = start;

    int skip_registers = 12;
    int max_registers = MODBUS_MAX_READ_REGISTERS;
    if (seg_do == segment || seg_di == segment) {
        max_registers = MODBUS_MAX_READ_BITS;
    }

    vector_t *tag_list = &device->_product->thing_model[segment];
    if (0 == vector_size(tag_list)) {
        return;
    }

    for (i = 0; running && i < vector_size(tag_list); ++i) {
        thing_model_t *tag = vector_get(tag_list, i);

        if (tag->catalog != catalog) {
            continue;
        }
        int curr = tag->offset;
        next = curr + get_data_len(tag->data_type) - 1;

        //第一个寄存器
        if (-1 == start) {
            start = curr;
            prev = next;
            continue;
        }

        /*if (!thing_model->_auto && thing_model->_start) {
         //不允许优化
         modbus_collect(device, segment, tag_list, start, prev - start + 1);
         start = curr; //下一个招测起点包含当前寄存器
         } else if (curr - prev > 1 && !thing_model->_skip) {
         //寄存器不相临 && 不允许跳过
         modbus_collect(device, segment, tag_list, start, prev - start + 1);
         start = curr; //下一个招测起点包含当前寄存器
         } else */if (curr - prev > skip_registers) {
            //超过要跳过的最大寄存器数量
            modbus_collect(channel, device, segment, tag_list, start, prev - start + 1);
            start = curr; //下一个招测起点包含当前寄存器
        } else if (next - start >= max_registers) {
            //超过单条指令的最大寄存器数量
            modbus_collect(channel, device, segment, tag_list, start, prev - start + 1);
            start = curr; //下一个招测起点包含当前寄存器
        }

        prev = next;
    }

    if (-1 != start) {
        modbus_collect(channel, device, segment, tag_list, start, prev - start + 1);
    }
}

void sched_modbus_device(channel_t *channel, data_catalog_e catalog, device_t *device) {
    int rc, i;
    segment_e segments[] = {seg_do, seg_di, seg_ao, seg_ai};

    if (!protocol_is_modbus(device->_product->protocol)) {
        return;
    }

    if (!is_retry_sched(device)) {
        return;
    }

    rc = modbus_init(channel, device);
    if (rc < 0) {
        log_error("failed to initialize(u0|ubb|i0|ibb|*digits) for modbus %d/%s", device->super.id, device->super.name);
        return;
    }

    for (i = 0; running && i < sizeof(segments) / sizeof(segments[0]); ++i) {
        sched_modbus_device_segment(channel, catalog, device, segments[i]);
    }
}

//判断通道链路是否存在畅通
int judge_channel(channel_t *channel) {
    for (int i = 0; running && i < vector_size(&channel->devices); ++i) {
        device_t *device = vector_get(&channel->devices, i);
        if (device->_stat.successful > 0) {
            return 1;
        } else if (device->_stat.total_messages == 0) {
            return -1;
        }
    }
    return 0;
}

void sched_modbus_channel(channel_t *channel, data_catalog_e catalog) {
    int i;

    for (i = 0; running && i < vector_size(&channel->devices); ++i) {
        device_t *device = vector_get(&channel->devices, i);
        //轮询清零
        device->_stat.successful = 0;
        device->_stat.failed = 0;
        device->_stat.total_messages = 0;
        sched_modbus_device(channel, catalog, device);
    }
    if (judge_channel(channel) == 1) {
        channel->_stat.retries = 0;
    } else if (judge_channel(channel) == 0) {
        channel->_stat.retries++;
    }
}

void sched_modbus_catalogs(channel_t *channel, seqiot_t *seqiot) {
    int i;

    struct timeval _tv;
    gettimeofday(&_tv, NULL);

    channel->_stat.total_messages = 0;
    channel->_stat.failed = 0;
    channel->_stat.successful = 0;

    for (i = 0; running && i < vector_size(&seqiot->catalogs); ++i) {
        data_catalog_t *catalog = vector_get(&seqiot->catalogs, i);

        if (catalog->_next_sched && catalog->_next_sched <= _tv.tv_sec) {
            catalog->_next_sched = next_sched(_tv.tv_sec, &catalog->read);

            struct tm _tm;
            localtime_r(&catalog->_next_sched, &_tm);
            log_debug("sched=%04d-%02d-%02d %02d:%02d:%02d, catalog=%d, period=%d, delay=%d",
                      _tm.tm_year + 1900, _tm.tm_mon + 1, _tm.tm_mday, _tm.tm_hour, _tm.tm_min, _tm.tm_sec,
                      catalog->catalog, catalog->read.period, catalog->read.delay);

            sched_modbus_channel(channel, catalog->catalog);
        }
    }

    //channel通道超过重试次数代表通道不通，需要重启进程
    if (channel->_stat.retries >= 3) {
        running = false;
    }

    gettimeofday(&_tv, NULL);
    usleep(1000 * 1000L - _tv.tv_usec);
}

/*
static int mosquitto_check_rc(int rc, const char *method) {
    if (MOSQ_ERR_ERRNO == rc || MOSQ_ERR_KEEPALIVE == rc) {
        log_error("%s = %d, %s", method, errno, strerror(errno));
    } else if (MOSQ_ERR_SUCCESS != rc) {
        log_error("%s = %d, %s", method, rc, mosquitto_strerror(rc));
    }
    return rc;
}

typedef struct {
    uint16_t offset;
    uint16_t value;
    int channel_id;
    time_t timestamp;
} ctrl_t;

ctrl_t *ctrl_new() {
    return calloc(1, sizeof(ctrl_t));
}

void ctrl_free(ctrl_t *ptr) {
    if (!ptr) return;
    free(ptr);
}

int json_perse(const cJSON *json, ctrl_t *ctrl) {
    if (NULL == ctrl || NULL == json) return -1;

    cJSON *item = cJSON_GetArrayItem(json, 0);
    for (cJSON *next = item; next; next = next->next) {
        if (!strcmp("channel_id", next->string)) {
            ctrl->channel_id = cJSON_GetNumberValue(item);
        } else if (!strcmp("value", next->string)) {
            ctrl->value = cJSON_GetNumberValue(item);
        } else if (!strcmp("offset", next->string)) {
            ctrl->offset = cJSON_GetNumberValue(item);
        } else if (!strcmp("timestamp", next->string)) {
            ctrl->timestamp = cJSON_GetNumberValue(item);
        }
    }

    return 0;
}

//modbus写入寄存器部分，控制操作
static int modbus_write(modbus_t *modbus, uint16_t offset, uint16_t value) {
    int rc = 0;
    rc = modbus_write_registers(modbus, offset, 1, &value);
    if (rc < 0) {
        log_error("modbus写入出错：modbus_write_registers = %d\n", rc);
        return -1;
    }
    return rc;
}

static int _control_callback(struct mosquitto *mosquitto, void *userdata, const struct mosquitto_message *message) {
    //获取消息队列信息参数
    char *payload = message->payload;
    payload[message->payloadlen] = '\0';
    cJSON *json = cJSON_Parse(payload);
    channel_t *channel = userdata;
    ctrl_t *ctrl = ctrl_new();
    int rc = json_perse(json, ctrl);
    if (rc == 0)
        modbus_write(channel->modbus, ctrl->offset, ctrl->value);

    ctrl_free(ctrl);
    cJSON_free(json);
}

//控制部分
void sched_control(channel_t *channel, struct mosquitto *mosquitto) {
    int rc = mosquitto_subscribe(mosquitto, NULL, "/user/control", 1);
    if (MOSQ_ERR_SUCCESS == rc) {
        rc = mosquitto_subscribe_callback(_control_callback,
                                          channel,
                                          "/user/control",
                                          1,
                                          "127.0.0.1",
                                          1883,
                                          "dev_control",
                                          60,
                                          false,
                                          NULL,
                                          NULL,
                                          NULL,
                                          NULL);
        if (rc < 0) log_error("control_callback error!");
    }
}
*/

void sched_modbus(channel_t *channel, seqiot_t *seqiot) {
    int i;

    struct timeval _tv;
    gettimeofday(&_tv, NULL);
    //必须调用一次__timezone才有值
    localtime(&_tv.tv_sec);

    /* //modbus控制
     mosquitto_lib_init();
     struct mosquitto *mosquitto = mosquitto_new("mqtt_pub", false, NULL);
     int rc = mosquitto_connect(mosquitto, seqiot->local_address, 1883, 60);
     mosquitto_check_rc(rc, "mosquitto_connect");
 */
    for (i = 0; i < vector_size(&seqiot->catalogs); ++i) {
        data_catalog_t *catalog = vector_get(&seqiot->catalogs, i);
        if (catalog->read.period < 0) {
            catalog->_next_sched = _tv.tv_sec;
        } else {
            catalog->_next_sched = next_sched(_tv.tv_sec, &catalog->read);
        }
    }

    while (running) {
        sched_modbus_catalogs(channel, seqiot);
        /* rc = mosquitto_loop(mosquitto, -1, 10);
         mosquitto_check_rc(rc, "mosquitto_loop");
         if (MOSQ_ERR_NO_CONN == rc || MOSQ_ERR_CONN_LOST == rc) {
             rc = mosquitto_reconnect(mosquitto);
             mosquitto_check_rc(rc, "mosquitto_reconnect");
         }
         sched_control(channel, mosquitto);*/
    }

    /*mosquitto_disconnect(mosquitto);
    mosquitto_destroy(mosquitto);
    mosquitto_lib_cleanup();*/
}
